package com.atguigu.edu.realtime.regist.app;

import com.atguigu.edu.realtime.common.base.BaseSqlApp;
import com.atguigu.edu.realtime.common.constant.Constant;
import com.atguigu.edu.realtime.common.util.SqlUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class DwdUserRegistApp extends BaseSqlApp {

    public static void main(String[] args) {
        new DwdUserRegistApp().start(11202,4,"DwdUserRegistApp",Constant.TOPIC_DWD_USER_REGISTER);
    }

    @Override
    public void handle(StreamExecutionEnvironment env, StreamTableEnvironment tableEnv) {
        tableEnv.executeSql("create table topic_db(\n" +
                "\t`database` String,\n" +
                "\t`data` MAP<string,string>,\n" +
                "\t`old` MAP<string,string>,\n" +
                "\t`type` String,\n" +
                "\t`table` String,\n" +
                "\t`ts` bigint,\n" +
                "  pt as proctime(),\n" +
                "  et as TO_TIMESTAMP_LTZ(ts, 0),\n" +
                "  WATERMARK FOR et AS et \n" +
                ")" + SqlUtil.getKafkaDDL("topic_db","DwdUserRegistApp"));

    //过滤出注册的数据
    //    tableEnv.sqlQuery("select * from traffic_page").execute().print();
        Table table = tableEnv.sqlQuery("select \n" +
                "`data`['id'] user_id,\n"+
                " `data`['create_time'] regesit_time,\n" +
                "\t`data`['gender'] gender,\n" +
                "\t`data`['birthday'] birthday,\n" +
                "\t ts\n"+
                " from topic_db\n" +
                " where `table` = 'user_info'\n" +
                " and `type` = 'insert'");

        //table.execute().print();

        String sql = "create table user_regest(\n" +
                "\tuser_id String,\n" +
                "\tregesist_time String,\n" +
                "\tgender String,\n" +
                "\tbirthday String,\n" +
                "\tts bigINT,\n" +
                "PRIMARY KEY (user_id) NOT ENFORCED" +
                ")" + SqlUtil.getUpsertKafkaDDL(Constant.TOPIC_DWD_USER_REGISTER);
        tableEnv.executeSql(sql);

        table.executeInsert("user_regest");

    }

}
